package com.liyong.res.producer;

import com.liyong.res.util.KafkaConstant;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * create by yongli on 2020-05-17 14:52
 */

public class SimpleProducer {
    public static final Logger logger = LoggerFactory.getLogger(SimpleProducer.class);

    public static void send() {
        Properties props = new Properties();
        props.put("bootstrap.servers", KafkaConstant.KAFKA_SERVER_ADDRESS);
        props.put("acks", "0");
        props.put("retries", 1);
        props.put("batch.size", 16384);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props); // 这个producer包含了好多对象啊

        int count = 1000;
        for (int i = 0; i < count; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(
                    "topicDemo", "key2",
                    "I will go back, and give you a big gift");

//        // 不是大名鼎鼎的callable接口
//        Callback callback = (metadata, exception) -> {
//            if (exception != null) {
//                logger.error("Encounter exception " + exception);
//            }else {
//                logger.info("send success");
//            }
//        };

            // 返回一个RecordMetadata数据,虽然是异步的，但是要有返回的状态信息
            producer.send(record);
        }
    }

}

